package org.mashart.pink.m.ws;

import java.util.Vector;
import java.util.Timer;
import java.util.TimerTask;
//import java.io.OutputStream;
//import java.io.IOException;
import javax.microedition.rms.RecordStore;
import javax.microedition.rms.RecordStoreException;
import javax.microedition.rms.RecordEnumeration;
import org.mashart.pink.m.gps.Waypoint;
import org.mashart.pink.m.store.FileManager;
import org.mashart.pink.m.store.Preferences;

/**
 * Restful Client with fallback mechanism.
 * It depends on both RestfulClient and FileConnector.
 * Buffer all the incoming messages in record store, and then send via HTTP POST (by RestfulClient).
 * When terminating, first try to send the remaining buffered messages;
 * if connection fails, save them as a file on mobile phone via FileConnector.
 */
public class RobustRestfulClient implements WsConnector {
    /**
     * Max length of a action string
     */
    private final static int MAX_LENGTH_ACTION = 20;
    /**
     * RestfulClient, can send the messages by HTTP POST
     */
    private RestfulClient rc;
    /**
     * size of in buffer
     */
    private final int sizeInbuf = 64;
    /**
     * buffer of incoming messages
     */
    private String [] inbuf;
    /**
     * mark of if the message in buffer is an action
     */
    private boolean [] isAct;
    /**
     * mark of if the message in buffer is valid
     */
    private boolean [] valid;
    /**
     * smallest unsent token number of the messages, i.e, the left boundary of the receive window
     */
    private int k;
    /**
     * batch vector
     */
    private Vector batch;
    /**
     * max size of batch
     */
    private final int maxSizeBatch = 16;
    /**
     * max size of out buffer
     */
    private final int maxSizeOutbuf = 64;
    /**
     * out buffer
     */
    private Vector outbuf;
    /**
     * record id of the messages in output buffer
     */
//    private int [] recordIdOutbuf;
    /**
     * pointer of first unsent record id
     */
//    private int ptUnsentRecordId;
    /**
     * max number of unprocessed messages
     */
    private final int MAX_UNPROCCESSED = 128;
    /**
     * queue of outgoing messages
     */
    private RecordStore outrs;
    /**
     * switch of the working status
     */
    private boolean moving;
    /**
     * switch of the working status
     */
    private boolean sending;
    /**
     * semaphore for multithreads synchronization
     */
    private final Object semMove;
    private final Object semSend;
    /**
     * token for arriving push thread
     */
    private int tokenArrival;
    /**
     * token of processing push thread
     */
    private int numProcessed;
    /**
     * Timer
     */
    Timer timerMove;
    Timer timerSend;
    /**
     * timer task
     */
    TimerTask taskMove;
    TimerTask taskSend;
    /**
     * interval of sending
     */
    private int intervalSend;
    /**
     * interval of moving
     */
    private int intervalMove;
    /**
     * if the move timer is working on slower rate
     */
    private boolean moveSlow;

    // debug monitor variable
    private int numMoved;
    private int numSent;
    private int numSendFailed;
    private int numInbuf;
    private int sizeOutrs;
    private int sizeOutbuf;

    /**
     * Constructor
     */
    public RobustRestfulClient() {
        Preferences pref = Preferences.getInstance();
        String strInterval = pref.get(Preferences.kSendInterval);
        if ( strInterval != null ){
            intervalSend = Integer.parseInt(strInterval) * 1000;
        }else{
            intervalSend = 5000;
        }
        // when the send is ready, need to move something
        // when the in buffer is full, also need to move something
        // so initialize interval for moving as the smaller of sending interval and size of in buf
        intervalMove = Math.min(intervalSend, sizeInbuf * 1000);
        
        semMove = new Object();
        semSend = new Object();
        rc = new RestfulClient();
//        rc = new FakeRestfulClient();
        try {
            outrs = RecordStore.openRecordStore("outrs", true);
        } catch (RecordStoreException rse) {
            rse.printStackTrace();
        }
        
        inbuf = new String [sizeInbuf];
        isAct = new boolean [sizeInbuf];
        valid = new boolean [sizeInbuf];
        
        for( int i = 0 ; i < sizeInbuf ; i++ ){
            valid[i] = false;
        }
        k = 0;

        batch = new Vector(maxSizeBatch);

        outbuf = new Vector(maxSizeOutbuf);
//        recordIdOutbuf =  new int [maxSizeOutbuf];
//        ptUnsentRecordId = maxSizeOutbuf;
    }

    /**
     * Configure
     * @return 1 if succeeded, -1 otherwise.
     */
    public int configure() {
        rc.configure();
        return 1;
    }

    /**
     * Initialize the record store.
     * @return 1 if succeeded, -1 otherwise.
     */
    public int initialize() {
        rc.initialize();

        numMoved = 0;
        numSent = 0;
        numSendFailed = 0;
        numInbuf = 0;
        sizeOutrs = 0;
        sizeOutbuf = 0;
        moving = true;
        sending = true;

        // send all the remainning data, if any
        sendAll();
//        save();

        timerMove = new Timer();
        taskMove =
                new TimerTask() {
                    public void run() {
                        synchronized (semMove) {
                            if (moving) {
                                move();
                            }
                        }
                    }
                };
        timerMove.schedule(taskMove, 0, intervalMove);
        moveSlow = false;

        timerSend = new Timer();
        taskSend =
                new TimerTask() {
                    public void run() {
                        synchronized (semSend) {
                            if (sending) {
                                send();
                            }
                        }
                    }
                };
        timerSend.schedule(taskSend, intervalSend / 2, intervalSend);
        return 1;
    }

    /**
     * Get the status
     * Debug use
     */
    public String status() {
        StringBuffer sb = new StringBuffer();

        sb.append("Arrrived/Processed/Unprocessed: ").append(tokenArrival).append('/').append(numProcessed).append('/')
                .append(tokenArrival-numProcessed).append('\n');
        sb.append("moved/sent succ/fail: ").append(numMoved).append('/').append(numSent).append('/').append(numSendFailed).append('\n');
        sb.append("inbuf/outrs/outbuf: ").append(numInbuf).append('/').append(sizeOutrs).append('/').append(sizeOutbuf).append('\n');
        sb.append("free mem:").append(Runtime.getRuntime().freeMemory()).append('\n');

        return sb.toString();
    }

    /**
     * Move the message from inrs to outrs
     */
    public void move() {
        // header position in buffer
        int header;
        // if the unprocesssed messages become too many
        if ( tokenArrival - k > MAX_UNPROCCESSED ) {
            // give up waiting for k and successive unreceived messages, until the first valid message
            // if there is not any valid message in buffer, skip "sizeInbuf" of messages
            for ( int i = 0 ; i < sizeInbuf && !valid[ k % sizeInbuf ]; i++ ){
                k++;
            }
        }

        header = k % sizeInbuf;
        // while the header message is valid
        while ( valid[ header ] ) {
            // get the number of successive waypoint messages at header
            int n = 0;
            int p = ( k + n ) % sizeInbuf;
            while( n < sizeInbuf && valid[ p ] && !isAct[ p ] ){
                // counter the number of successive waypoint messages
                n++;
                p = ( k + n ) % sizeInbuf;
            }
//            System.out.println("k=" + k + ",header=" + header + ",n=" + n);

            // if there are more than one waypoints at header
            if ( n > 1 ) {
//                // DEBUG
//                for(int i = 0 ; i < sizeInbuf ; i++){
//                    if( !valid[i] ){
//                        System.out.println("inbuf[" + i + "]=" + " invalid");
//                    }else{
//                        System.out.println("inbuf[" + i + "]=" + inbuf[i]);
//                    }
//                }

                for ( int i = 0 ; i < maxSizeBatch && i < n ; i++ ) {
                    // DEBUG
//                    System.out.print("[" + (System.currentTimeMillis() % 1000000) + "]");
//                    System.out.print("moving:[" + header + "]");
//                    System.out.println(inbuf[ header ]);

                    batch.addElement(inbuf[ header ]);
                    valid[ header ] = false;
                    k++;
                    header = k % sizeInbuf;
                    numMoved++;
                    numInbuf--;
                }
                String strBatch = Waypoint.aggregateHttpQuery(batch);
                batch.removeAllElements();
                moveToOutrs(strBatch);
            }else if ( n == 1 ) {
                // DEBUG
//                System.out.print("[" + (System.currentTimeMillis() % 1000000) + "]");
//                System.out.print("moving:");
//                System.out.println(inbuf[header]);
                
                moveToOutrs(inbuf[header]);
                valid[header] = false;
                k++;
                header = k % sizeInbuf;
                numMoved++;
                numInbuf--;
            }else {
                // DEBUG
//                System.out.print("[" + (System.currentTimeMillis() % 1000000) + "]");
//                System.out.print("moving:");
//                System.out.print("act:");
//                System.out.println(inbuf[header]);

                // if the header is valid, and there is no waypoint at header
                // the header must be action
                moveToOutrs(inbuf[header]);
                valid[header] = false;
                k++;
                header = k % sizeInbuf;
                numMoved++;
                numInbuf--;
            }
        }
    }

    /**
     * move a message into outrs
     */
    public void moveToOutrs( String msg ){
        sizeOutrs++;
        byte[]b = msg.getBytes();
        synchronized ( outrs ) {
            try{
                outrs.addRecord(b, 0, b.length);
            }catch (RecordStoreException rse) {
            }
        }
    }

    /**
     * sena all the messages
     */
    private int sendAll(){
        int res = 1;
        // wait until outrs is cleared
        synchronized ( outrs ) {
            try{
                while ( outrs.getNumRecords() > 0 ) {
//                    System.out.println("waiting to quit");
                    res = send();
                    if(res < 0){
                        break;
                    }
                }
            }catch (RecordStoreException rse) {
            }
        }
        return res;
    }

    /**
     * Send the message in the output record store
     * @return
     */
    private int send() {
        // TODO delete only after send successfully
        // move from outrs to outbuf
        moveToOutbuf();
        // send from outbuf
        return sendOutbuf();
        // remove the sent records
//        removeRecordsOutrs();
    }

    private void moveToOutbuf(){
        RecordEnumeration re;
        // recordId
        int id;
        byte[] b;
        String q;

//        int size = outbuf.size();
//        if ( ptUnsentRecordId == maxSizeOutbuf ) {
            
            synchronized( outrs ){
                try {
                    // switch the mover back to normal, when send succeeded last time
                    // and the buffered waypoints are below a threshold
                    if ( outrs.getNumRecords() < 4 ) {
                        switchMoverSlow(false);
                    }

                    re = outrs.enumerateRecords(null, null, false);

                    while ( outbuf.size() < maxSizeOutbuf && re.hasNextElement() ) {
                        id = re.nextRecordId();
                        b = outrs.getRecord(id);
                        q = new String(b);
                        outbuf.addElement(q);
                        outrs.deleteRecord(id);
//                        outbuf [ i ] = q;
//                        recordIdOutbuf [ i ] = id;
                        sizeOutbuf++;
                        sizeOutrs--;
                    }
                } catch (RecordStoreException rse) {
                }
            }
//        }else{
//            // switch the mover to slow, once failed to send last time
//            switchMoverSlow(true);
//        }
    }
    
    private void switchMoverSlow(boolean toSlow){
        // if current status is what we ask for
        if ( moveSlow == toSlow ) {
            return;
        }

        // get the normal rate
        long period = intervalMove;
        // switch to slow
        if ( !moveSlow && toSlow ) {
            period = maxSizeBatch * 1000;
            moveSlow = true;
//            System.out.println("switch mover slow");
        }
        // swtich to normal
        else if ( moveSlow && !toSlow ) {
            period = intervalMove;
            moveSlow = false;
//            System.out.println("switch mover normal");
        }

        if ( timerMove != null ) {
            timerMove.cancel();
            timerMove = null;
            timerMove = new Timer();
            waitTilFinishMove();
            timerMove.schedule(taskMove, 0, period);
        }
    }

//    private void removeRecordsOutrs(){
//        // recordId
//        int id;
//
//        synchronized( outrs ){
//            for ( int i = 0 ; i < ptUnsentRecordId && outbuf[i] != null ; i++ ) {
//                id = recordIdOutbuf [ i ];
//                try {
//                    outrs.deleteRecord( id );
//                } catch (RecordStoreException rse) {
//                }
//                outbuf[i] = null;
//    //                    System.out.println("deleting outrs id="+id);
//                sizeOutrs--;
//            }
//        }
//    }

    /**
     * Send from out buffer
     */
    private int sendOutbuf(){
        int res = 1;
        String q;
        while( !outbuf.isEmpty() ){
            q = (String) outbuf.elementAt(0);
            if(q == null || q.length() == 0){
                outbuf.removeElementAt(0);
                sizeOutbuf--;
                continue;
            }else if( q.length() <= MAX_LENGTH_ACTION ) {
                // get the action name
                String action = q.substring("action=".length());
                // pushAction
                res = rc.pushAction(action);
            } else {
                res = rc.push(q);
            }
                    // delete the record that is sent successfully
                System.out.println("sending");
            q = null;

//            System.out.println("sending id="+id.intValue());
            if (res > 0) {
                    // delete the record that is sent successfully
                System.out.println("send succ");
                outbuf.removeElementAt(0);
                sizeOutbuf--;
                numSent++;
            } else {
                    // delete the record that is sent successfully
                System.out.println("send fail");
                // switch the mover to slow, once failed to send last time
                switchMoverSlow(true);
                numSendFailed++;
                // jump out when failed
                break;
            }
        }
        return res;
    }

    /**
     * Push the string into buffer. It will be sent later.
     * When the buffer is full, discard some old elements until there is one vacancy.
     * @param s
     * @return
     */
    public int push(final String s) {

        // DEBUG
//        long debugTimer = System.currentTimeMillis() % 1000000;
//        final String s = "T=" + Long.toString(debugTimer) + "&n=" + tokenArrival +"&" + sDebug;
//        final String s = "n=" + tokenArrival +"&" + sDebug;

        // assign an arrival token for each thread
        final int tk = tokenArrival++;

        // make a new thread to handle the push
        new PushThread(tk, s, false).start();
        return 1;
    }

    /**
     * Push the action string into buffer. It will be sent later.
     * When the buffer is full, discard some old elements until there is one vacancy.
     * @param s
     * @return
     */
    public int pushAction(final String s) {

        // DEBUG
//        long debugTimer = System.currentTimeMillis() % 1000000;
//        final String s = "T=" + Long.toString(debugTimer) + "&n=" + tokenArrival +"&" + sDebug;
//        final String s = "n=" + tokenArrival +"&" + sDebug;

        // assign an arrival token for each thread
        final int tk = tokenArrival++;

        // make a new thread to handle the push
        new PushThread(tk, "action=" + s, true).start();
        return 1;
    }

    class PushThread extends Thread {

        final private int token;
        final private String msg;
        final private boolean isAction;

        PushThread(int tk, String s, boolean isAct) {
            super();
            this.token = tk;
            this.isAction = isAct;
            this.msg = s;
            
//            if ( s.length() <= MAX_LENGTH_ACTION ) {
//                this.msg = s;
//            }else{
//                this.msg = "n=" + tk + "&" + s;
//            }
        }

        public void run() {
            // if it is new
            if ( token >= k ){
                // wait until its turn
                while ( token >= k + sizeInbuf ) {
                    try{
                        Thread.sleep( 1000 );
                    }catch(InterruptedException ex){
                    }
                }


                // position in buffer
                int pos = token % sizeInbuf;
                inbuf [ pos ] = msg;
                isAct [ pos ] = isAction;
                valid [ pos ] = true;
                numInbuf++;
                numProcessed++;

                // DEBUG
//                System.out.print("[" + (System.currentTimeMillis() % 1000000) + "]");
//                System.out.print("pushing:[" + pos + "]");
//                System.out.println(msg);
            }
        }
    }

    /**
     * Terminate the connector.
     * Save the unsent data as a file.
     */
    public int terminate() {
        // wait until all push threads are processed
        while ( k < tokenArrival ) {
//                    System.out.println("waiting to quit");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
        }

        // stop the mover
        timerMove.cancel();
        timerMove = null;
        moving = false;
        // wait til move stop
        waitTilFinishMove();
        // ensure all are moved
        move();

//        int num;
//        // wait until outrs is cleared
//        synchronized ( outrs ) {
//            try{
//                while ( (num = outrs.getNumRecords()) > 0 ) {
////                    System.out.println("waiting to quit");
//                    try {
//                        // wait num/2 seconds, because one HTTP session takes about 0.5 second
//                        outrs.wait(num * 500);
//                    } catch (InterruptedException e) {
//                    }
//                }
//            }catch (RecordStoreException rse) {
//            }
//        }
//        System.out.println("going to quit");

        // stop the sender
        timerSend.cancel();
        timerSend = null;
        sending = false;
        waitTilFinishSend();
        sendAll();

//        save();
        // termainte the RestfulClient
        rc.terminate();
        return 1;
    }

//    /**
//     * Wait until the move-send thread finishes
//     */
//    public void waitTilFinish() {
//        synchronized (semMove) {
//        }
//        synchronized (semSend) {
//        }
//    }

    public void waitTilFinishMove() {
        synchronized (semMove) {
        }
    }

    public void waitTilFinishSend() {
        synchronized (semSend) {
        }
    }

//    /**
//     * Save the unsent message in record store into a file, and delete the record store
//     */
//    private int save(){
//        int res = 1;
//        try{
//            if( outrs.getNumRecords() > 0 ){
//                System.out.println("saving remaining data");
//                FileManager fileManager = FileManager.getInstance();
//                // initialized the file manager
//                if ( FileManager.isAPIAvailable() && fileManager.initialize() > 0)
//                {
//                    // save the RecordStore
//                    res = fileManager.save(outrs);
//                    // clear ours
////                    clear(outrs);
//                    System.out.println("saved");
//                }else{
//                    res = -1;
//                }
//            }
//        }catch(RecordStoreException rse){
//            rse.printStackTrace();
//            res = -1;
//        }
//        return res;
//    }
//
//    /**
//     * Clear the queue
//     */
//    private void clear(RecordStore rs){
//        try{
//            RecordEnumeration re;
//            re = rs.enumerateRecords(null, null, false);
//            while( re.hasNextElement() ){
//                rs.deleteRecord(re.nextRecordId());
//            }
//        }catch(RecordStoreException rse){
//            rse.printStackTrace();
//        }
//    }

}